﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Runtime.Serialization;
using System.ServiceModel;
using System.ServiceModel.Description;
using System.Text;
using System.IO;
using System.Reflection;

namespace Paxos
{
    [DataContract(Name="Server GUID")]
    public struct SGUID: IComparable<SGUID>
    {
        [DataMember]
        public int Port;

        public int CompareTo(SGUID sguid)
        {
            if (Port > sguid.Port)
                return -1;
            else if (Port == sguid.Port)
                return 0;
            else
                return 1;
        }

        public override string ToString()
        {
            return Port.ToString();
        }

        public string GetProposerEndpointAddress()
        {
            return String.Format("net.tcp://localhost:{0}/proposer", Port+100);
        }
    
        public string GetAcceptorEndpointAddress()
        {
            return String.Format("net.tcp://localhost:{0}/acceptor", Port+200);
        }
    }

    public static class PaxosHelper
    {
        public static string JoinToString(this IEnumerable<SGUID> list)
        {
            return String.Join(",", list.ToArray());
        }
    }

    [DataContract]
    public class ServerView
    {
        [DataMember]
        public List<SGUID> Servers;

        [DataMember]
        public SGUID Primary;

        public override string ToString()
        {
            return String.Format("{0} : {1}", Primary, String.Join(",", Servers));
        }
    }
    
    [DataContract]
    public enum IssueResult
    {
        Success = 0,
        NoConsensus,
        OutOfDate
    };

    [ServiceContract]
    public interface IPaxosProproser
    {
        [OperationContract]
        void HitHeart();

        [OperationContract]
        ServerView GetView();

        [OperationContract]
        IssueResult IssueCommand(PaxosCommand command);

        [OperationContract]
        PaxosCommand[] GetLatest(long startBallot);
    }

    [DataContract]
    public class PromiseResult
    {
        [DataMember]
        public bool Answer;

        [DataMember]
        public SelfAppointProposal Promised;

        [DataMember]
        public ConcreteProposal Accepted;

        public override string ToString()
        {
            return String.Format("{0} Pd: {1} Ad: {2}", Answer, Promised, Accepted);
        }
    }

    [DataContract]
    public class AcceptResult
    {
        [DataMember]
        public bool Answer;

        [DataMember]
        public SelfAppointProposal Promised;

        public override string ToString()
        {
            return String.Format("{0} Pd: {1}", Answer, Promised);
        }
    }

    [ServiceContract]
    public interface IPaxosAcceptor
    {
        [OperationContract]
        PromiseResult Promise(SelfAppointProposal proposal);

        [OperationContract]
        AcceptResult Accept(ConcreteProposal proposal);

        [OperationContract]
        void Commit(ConcreteProposal proposal);

        [OperationContract]
        void CatchUp();
    }

    [DataContract]
    [KnownType("GetKnownCommands")]
    public abstract class PaxosCommand
    {
        private static Type[] commands;

        public abstract void Commit(ReplicaStateMachine state);

        private static Type[] GetKnownCommands()
        {
            if (commands == null)
            {
                Console.WriteLine("Call GetKnownCommands");
                commands = (from assembly in AppDomain.CurrentDomain.GetAssemblies()
                            from t in assembly.GetExportedTypes()
                            where t.IsClass && t.IsSubclassOf(typeof(PaxosCommand))
                            select t).ToArray();
                if (commands != null && commands.Length > 0)
                {
                    Console.WriteLine("Command Count: {0}", commands.Length);
                    foreach (Type t in commands)
                    {
                        Console.WriteLine(t.Name);
                    }
                }
            }

            return commands;                
        }
    }

    [DataContract]
    [KnownType(typeof(ConcreteProposal))]
    [KnownType(typeof(SelfAppointProposal))]
    [KnownType(typeof(EmptyProposal))]
    public abstract class PaxosProposal : IComparable<PaxosProposal>
    {
        public static EmptyProposal EMPTYPROPOSAL = new EmptyProposal();
        
        /// <summary>
        /// the round of paxos
        /// </summary>
        [DataMember]
        public long Ballot;

        /// <summary>
        /// the proposal ID
        /// </summary>
        [DataMember]
        public long ID;

        [DataMember]
        public SGUID DecreeGuid;

        public int CompareTo(PaxosProposal p)
        {
            if (p == null)
            {
                return 1;
            }

            if (Ballot == p.Ballot)
            {
                if (ID == p.ID)
                {
                    return DecreeGuid.CompareTo(p.DecreeGuid);
                    //if (DecreeGuid == null && p.DecreeGuid == null)
                    //{
                    //    return 0;
                    //}
                    //else if (DecreeGuid != null && p.DecreeGuid == null)
                    //{
                    //    return 1;
                    //}
                    //else if (DecreeGuid == null && p.DecreeGuid != null)
                    //{
                    //    return -1;
                    //}
                    //else
                    //{
                    //    return DecreeGuid.Value.CompareTo(p.DecreeGuid.Value);
                    //}
                }
                else if (ID > p.ID)
                {
                    return 1;
                }
                else
                {
                    return -1;
                }
            }
            else if (Ballot > p.Ballot)
            {
                return 1;
            }
            else
            {
                return -1;
            }
        }

        public override string ToString()
        {
            return String.Format("Proposal {0} {1} {2} {3}", this.GetType().Name, Ballot, ID, DecreeGuid);
        }
    }

    [DataContract]
    public sealed class SelfAppointProposal : PaxosProposal
    {
    }
    
    [DataContract]
    public sealed class EmptyProposal : PaxosProposal
    {
    }

    [DataContract]
    public sealed class ConcreteProposal : PaxosProposal
    {
        [DataMember]
        public PaxosCommand Command;

        public override string ToString()
        {
            return String.Format("Proposal {0} {1} {2} {3} {4}", this.GetType().Name, Ballot, ID, DecreeGuid, Command);
        }
    }

    public enum ReplicaState { RSActive, RSManager, RSGuest };

    public class Replica
    {
        public ReplicaState State;
        public ServerView CurrentView;
        public long LatestCommandID;
        public ServerView AcceptedView;
    }

    public interface IReplicaStateMachine
    {
        void LogCommand(PaxosCommand command);
    }

    public class CommandLogger
    {
        private DataContractSerializer dcs = new DataContractSerializer(typeof(PaxosCommand));
        private string _filename;

        public bool CanReplay { private set; get;}

        public CommandLogger(string filename)
        {
            _filename = filename;
            CanReplay = true;
        }

        public void Replay(ReplicaStateMachine state)
        {
            if (!CanReplay)
            {
                return;
            }

            if (File.Exists(_filename))
            {
                byte[] commandBuffer = new byte[4096];
                using(FileStream inStream = new FileStream(_filename, FileMode.Open, FileAccess.Read))
                using(BinaryReader reader= new BinaryReader(inStream))
                {
                    while (inStream.Position < inStream.Length)
                    {
                        // read a command log
                        int length = reader.ReadInt32();
                        MemoryStream mmStream = new MemoryStream(commandBuffer, 0, length);
                        int offset = 0;
                        while (length > 0)
                        {
                            int readCount = reader.Read(commandBuffer, offset, length);
                            offset += readCount;
                            length -= readCount;
                        }

                        // deserialize to command
                        PaxosCommand command = dcs.ReadObject(mmStream) as PaxosCommand;

                        // execute
                        command.Commit(state);
                    }
                }
            }

            CanReplay = false;
        }

        public void Log(PaxosCommand command)
        {
            CanReplay = false;

            using (MemoryStream mmStream = new MemoryStream())
            using (FileStream outStream = new FileStream(_filename, FileMode.OpenOrCreate, FileAccess.Write))
            using (BinaryWriter writer = new BinaryWriter(outStream))
            {
                dcs.WriteObject(mmStream, command);
                mmStream.Position = 0;

                outStream.Seek(0, SeekOrigin.End);
                writer.Write((Int32)mmStream.Length);
                mmStream.CopyTo(outStream);
                outStream.Flush();
            }           
        }

        public PaxosCommand[] GetLatest(long startBallot)
        {
            List<PaxosCommand> ret = new List<PaxosCommand>();
            
            if (File.Exists(_filename))
            {
                byte[] commandBuffer = new byte[4096];
                using (FileStream inStream = new FileStream(_filename, FileMode.Open, FileAccess.Read))
                using (BinaryReader reader = new BinaryReader(inStream))
                {
                    // seek to start command
                    int length = 0;
                    for (int i = 0; i < startBallot; ++i)
                    {
                        length = reader.ReadInt32();
                        inStream.Position += length;
                    }

                    while(inStream.Position < inStream.Length)
                    {
                        length = reader.ReadInt32();
                        MemoryStream mmStream = new MemoryStream(commandBuffer, 0, length);
                        int offset = 0;
                        while (length > 0)
                        {
                            int readCount = reader.Read(commandBuffer, offset, length);
                            offset += readCount;
                            length -= readCount;
                        }
                        // deserialize to command
                        PaxosCommand command = dcs.ReadObject(mmStream) as PaxosCommand;
                        ret.Add(command);
                    }
                }
            }

            return ret.ToArray();
        }
    }
}
